package com.atguigu.flink.streamapi.transform;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Created by Smexy on 2022/11/21
 *
 *      在主流中，打印s1传感器，在侧流(支流)中，一个打印s3，一个打印s2
 */
public class Demo12_SideOutPut
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //OutputTag,必须用两个参数的构造器，显式指定流中数据的类型的TypeInformation
        OutputTag s3 = new OutputTag("s3", TypeInformation.of(WaterSensor.class));
        OutputTag s2 = new OutputTag("s2",TypeInformation.of(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .process(new ProcessFunction<WaterSensor, WaterSensor>()
            {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                    String id = value.getId();

                    if ("s1".equals(id)) {
                        out.collect(value);
                    } else if ("s2".equals(id)) {
                       /*
                            使用侧流输出

                            <X> void output(OutputTag<X> outputTag, X value)
                                OutputTag: 输出标记。标识这个流中输出的是什么，什么类型
                        */
                        ctx.output(s2, value);
                    } else if ("s3".equals(id)) {
                        ctx.output(s3, value);
                    }

                }
            });

        //获取主流打印
        ds.printToErr("主流s1");
        //获取支流（侧流）
        ds.getSideOutput(s2).print("侧流s2");
        ds.getSideOutput(s3).print("侧流s3");

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
